//
// Created by 王筱童 on 2022/5/14.
//

#include "slave_manager.h"
#include "redis_server.h"
#include "others.h"
#include <iostream>

asio::awaitable<void> slave_manager::sync(std::size_t index, const std::string &buf) {
        std::string data= serialize(buf);
#ifdef TEST
        std::cout<<"SYNC BUF "<<buf<<std::endl;
        std::cout<<"SYNC DATA "<<data<<std::endl;
#endif
        co_await asio::async_write(*socks[index].second,asio::buffer(data),asio::use_awaitable);
        std::string head(sizeof(std::uint64_t),'\0');
        co_await asio::async_read(*socks[index].second,asio::buffer(head),asio::use_awaitable);
        auto [len,_]=deserialize_head(head);
        head.resize(len,'\0');
        co_await asio::async_read(*socks[index].second,asio::buffer(head),asio::use_awaitable);
#ifdef TEST
        std::cout<<"SYNC RECV "<<head<<std::endl;
#endif
}

void slave_manager::add_slave(asio::ip::tcp::socket &&socket,const std::string& ip_port, const std::string &buf) {
    sock_ptr ptr=std::make_unique<asio::ip::tcp::socket>(std::move(socket));
    buf_sock data={ip_port,std::move(ptr)};
    socks[sock_indexs++]=std::move(data);
    //asio::co_spawn(server->run_context, sync(sock_indexs-1,buf),asio::detached);
}

void slave_manager::async_sync(const std::string &command, const std::vector<std::string> &args, std::size_t db_index) {
    for (int i = 0; i < socks.size(); ++i) {
        std::string data=command;
        for(auto &str:args){
            data+=' ';
            data+=str;
        }
        data+='\n';
        asio::co_spawn(server->run_context, sync_data(i,std::move(data),db_index),asio::detached);
    }
}

asio::awaitable<void> slave_manager::sync_data(std::size_t index, const std::string buf, const std::size_t db_index) {
    std::string res="SELECT "+std::to_string(db_index)+"\n";
    try {
        co_await sync(index,res);
        co_await sync(index,buf);}
    catch (std::exception&ec){
        socks.erase(index);
    }
}

slave_manager::slave_manager(redis_server *_server):server(_server) {

}

std::vector<std::pair<std::string,std::uint16_t>> slave_manager::get_all_slave_ip_port()const {
    std::vector<std::pair<std::string,std::uint16_t>> res;
    for(auto &conn:socks){
        std::pair<std::string,std::uint16_t> pair;
        auto tmp= split(conn.second.first,' ');
        pair.first=tmp.front();
        pair.second= convert<std::uint16_t>(tmp.back());
        res.push_back(std::move(pair));
    }
    return res;
}
